package com.hrt.datasender.datasent.socket;

import java.net.InetSocketAddress;
import java.util.Date;
import java.util.concurrent.Executors;

import com.hrt.datasender.web.operInfo.OperInfoService;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

import com.btcode.common.MyLog;
import com.btcode.log.ILog;
import com.hrt.datasender.datasent.service.handler.MessageDecode;
import com.hrt.datasender.datasent.service.handler.MessageReceiveProcess;

public class SocketClient {

    private ILog log;
    private ClientBootstrap bootstrap;
    private ChannelFuture connectFeature;
    private Date lastHeartBreakTime;
    
    private final int millTrans = 1000;
    private final int heartBreakInterval = 90;
    private String ip;
    private int port;
    private String clientId;// 连接标识，ip_端口
    private boolean isRestartting = false; // 是否正在重启中

    public SocketClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
        clientId = ip + "_" + port;
        log = MyLog.getInstance().getLogger(getClass().getName() + "_" + clientId);
        updateLastHeartBreakTime();

        // 启动消费队列任务
        ReconnectJob executer = new ReconnectJob();
        executer.start();
    }

    protected void startNetty() {
        NioClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(
                Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
        bootstrap = new ClientBootstrap(channelFactory);

        ChannelPipelineFactory channelPipelineFactory = new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                ChannelPipeline channelPipeline = Channels.pipeline();
                channelPipeline.addFirst("MessageDecode", new MessageDecode());
                channelPipeline.addLast("encode", new StringEncoder());
                channelPipeline.addLast("decode", new StringDecoder());
                channelPipeline.addLast("closeHandle", new ChannelCloseHandle());
                channelPipeline.addLast("MessageProcess", new MessageReceiveProcess(SocketClient.this));
                channelPipeline.addLast("log", new MessageLog());
                return channelPipeline;
            }
        };
        bootstrap.setOption("child.reuseAddress", true);
        bootstrap.setPipelineFactory(channelPipelineFactory);
        connect();
    }

    public void connect() {
        if (isRestartting()) {
            return;
        }
        log.info("连接服务器……");
        OperInfoService.getInstance().info("连接服务器……",ip,port);

        setRestartting(true);
        connectFeature = bootstrap.connect(new InetSocketAddress(ip, port));
        connectFeature.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture arg0) throws Exception {
                setRestartting(false);
                updateLastHeartBreakTime();
                if (arg0.isSuccess()) {
                    log.info("连接成功," + ip + ":" + port);
                    OperInfoService.getInstance().info("连接成功," + ip + ":" + port,ip,port);
                } 
                else {
                    log.info("连接失败," + ip + ":" + port);
                    OperInfoService.getInstance().info("连接失败," + ip + ":" + port,ip,port);
                }
            }
        });
    }

    /**
     * 心跳
     * 
     * @return
     */
    public boolean isActivite() {
        Date now = new Date();
        long timeDiff = (now.getTime() - lastHeartBreakTime.getTime()) / millTrans;

        if (timeDiff > heartBreakInterval) {
            return false;
        } 
        else {
            return true;
        }
    }

    public void updateLastHeartBreakTime() {
        this.lastHeartBreakTime = new Date();
    }

    public String getIp() {
        return ip;
    }

    public int getPort() {
        return port;
    }

    public String getClientId() {
        return clientId;
    }

    public boolean isRestartting() {
        log.info("获取isRestartting:" + isRestartting);
        return isRestartting;
    }

    private void setRestartting(boolean isRestartting) {
        log.info("设置isRestartting:" + isRestartting);
        this.isRestartting = isRestartting;
    }

    private class MessageLog extends SimpleChannelUpstreamHandler {

        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent msgEvent) throws Exception {
            super.messageReceived(ctx, msgEvent);
            log.info("收到消息：" + msgEvent.getMessage());
            OperInfoService.getInstance().info("收到消息：" + msgEvent.getMessage(),ip,port);
        }
    }

    public class ChannelCloseHandle extends SimpleChannelUpstreamHandler {
        @Override
        public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent msgEvent) throws Exception {
            super.channelDisconnected(ctx, msgEvent);
            String stateName = "";
            String value = "";
            if (msgEvent.getState() != null) {
                stateName = msgEvent.getState().name();
            }
            if (msgEvent.getValue() != null) {
                value = msgEvent.getValue().toString();
            }
            log.info("连接断开,State:" + stateName + "\nValue:" + value);
            OperInfoService.getInstance().info("连接断开,State:" + stateName + "\nValue:" + value,ip,port);
        }

        @Override
        public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent msgEvent) throws Exception {
            super.channelClosed(ctx, msgEvent);
            String stateName = "";
            String value = "";
            if (msgEvent.getState() != null) {
                stateName = msgEvent.getState().name();
            }
                
            if (msgEvent.getValue() != null) {
                value = msgEvent.getValue().toString();
            }
                
            log.info("连接关闭,State:" + stateName + "\nValue:" + value);
            OperInfoService.getInstance().info("连接关闭,State:" + stateName + "\nValue:" + value,ip,port);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext arg0, org.jboss.netty.channel.ExceptionEvent arg1)
                throws Exception {
            log.error(ip + port +"exceptionCaught连接出错了", arg1.getCause());
        }
    }

    /**
     * 判断客户端是否已准备好发送消息
     * 
     * @return
     */
    public boolean isClientPrepared() {
        if (connectFeature == null || connectFeature.getChannel() == null || !connectFeature.getChannel().isOpen()) {
            log.info("连接未准备好");
            OperInfoService.getInstance().info("连接未准备好",ip,port);
            return false;
        }
        if (!connectFeature.getChannel().isWritable()) {
            log.info("当前不可写");
            return false;
        }
        return true;
    }

    protected void writeMsg(Object msg) {

        ChannelFuture feature = connectFeature.getChannel().write(msg);

        feature.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture arg0) throws Exception {
                updateLastHeartBreakTime();
                if (!arg0.isSuccess()) {
                    if (arg0.getCause() != null) {
                        log.info("发送失败:" + arg0.getCause().getMessage() + "\r\n" + msg);
                        OperInfoService.getInstance().info("发送失败:" + arg0.getCause().getMessage() + "\r\n" + msg,ip,port);
                    }
                    else {
                        log.info("发送失败:" + msg);
                        OperInfoService.getInstance().info("发送失败:" + msg,ip,port);
                    }
                }
                else {
                    log.info("发送成功:" + msg);
                    OperInfoService.getInstance().info("发送成功:" + msg,ip,port);
                }
            }
        });
    }

    class ReconnectJob extends Thread {
        @Override
        public void run() {
            while (true) {
                if (!isActivite()) {
                    connect();
                }
                try {
                    Thread.sleep(millTrans);
                }
                catch (Exception e) {
                    log.error("重连线程ReconnectJob出错", e);
                }
            }
        }
    }

}
